package com.ek;

import java.io.Serializable;
import java.util.Hashtable;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ek.common.Bucket;
import com.ek.common.Context;
import com.ek.common.MessageHandler;
import com.ek.common.task.SendMulticastMessage;
import com.ek.kademlia.RoutingTableThreadSafe;
import com.ek.kademlia.handlers.DiscoveryMessageHandler;
import com.ek.kademlia.handlers.FindNodeRequestHandler;
import com.ek.kademlia.handlers.FindValueRequestHandler;
import com.ek.kademlia.handlers.StoreRequestHandler;
import com.ek.kademlia.task.FindValue;
import com.ek.kademlia.task.NodeLookup;
import com.ek.kademlia.task.StoreValue;
import com.ek.service.MulticastReceiver;
import com.ek.service.TaskManager;
import com.ek.service.mina.ComunicationService;
import com.ek.util.Utils;

public class Kademlia extends AbstractDht<String, Serializable> {
    private static final transient Logger log = LoggerFactory.getLogger(Kademlia.class);
    private byte[] id;
    private Context context = new Context();
    private boolean joined = false;

    public Kademlia() {
        init();
    }

    public Serializable get(Object key) {
        if (joined) {
            try {
                if (!(key instanceof String)) {
                    throw new IllegalArgumentException("the key type must be String.");
                }
                String keyString = (String) key;
                NodeLookup task = new NodeLookup(Utils.computeSHA1FromString(keyString), context);
                context.getTaskManager().put(task);
                List<Bucket> list = task.get();
                FindValue findValue = new FindValue(keyString, list, context);
                context.getTaskManager().put(findValue);
                List<Serializable> result = findValue.get();
                if (result.size() > 0)
                    return result.get(0);

            } catch (InterruptedException e) {
                log.error("get InterruptedException", e);

            } catch (ExecutionException e) {
                log.error("get ExecutionException", e);

            }
        } else {
            log.error("please first join into the network!");

        }
        return context.getData().get(key);
    }

    public Serializable put(String key, Serializable value) {
        if (joined) {
            try {
                NodeLookup task = new NodeLookup(Utils.computeSHA1FromString(key), context);
                context.getTaskManager().put(task);
                List<Bucket> list = task.get();
                StoreValue store = new StoreValue(key, value, list, context.getNodeID(), context.getComunicationService());
                context.getTaskManager().put(store);
                context.getData().put(key, value);

            } catch (InterruptedException e) {
                log.error("", e);

            } catch (ExecutionException e) {
                log.error("", e);

            }

        } else {
            log.error("please first join into the network!");

        }
        return null;
    }

    public void join() throws Exception {
        findNodes();
        joined = true;
    }

    public void leave() {
        context.getComunicationService().stop();
        context.getMulticastReceiver().stop();
        context.getTaskManager().stop();
        joined = false;
    }

    public byte[] getId() {
        return id;
    }

    private void findNodes() {
        context.getTaskManager().put(new SendMulticastMessage(context.getComunicationService()));
    }

    private void generateID() {
        log.info("Generating id ...");
        id = Utils.computeSHA1FromString(Utils.getLocalHostIP());
        log.info("Generated id:" + new String(id));
    }

    private void init() {
        log.info("init dht beans ...");

        generateID();
        context.setData(new Hashtable<String, Serializable>());
        context.setNodeID(id);
        context.setRoutingTable(new RoutingTableThreadSafe(id));
        context.setTaskManager(new TaskManager());
        context.setHandlers(getHandlers());

        ComunicationService comunicationService = null;
        try {
            comunicationService = new ComunicationService(context);

        } catch (Exception e) {
            log.error("Exception: init comunication service. ", e);

        }

        context.setComunicationService(comunicationService);
        context.setMulticastReceiver(new MulticastReceiver(context));

        log.info("start of services ...");
        ExecutorService pool = Executors.newFixedThreadPool(2);
        pool.execute(context.getTaskManager());
        pool.execute(context.getMulticastReceiver());
        log.info("started");

        log.info("init completed.");
    }

    private Hashtable<String, MessageHandler> getHandlers() {
        log.debug("init handlers");
        Hashtable<String, MessageHandler> handlers = new Hashtable<String, MessageHandler>();
        handlers.put(FindNodeRequestHandler.getMessageClass(), new FindNodeRequestHandler(context));
        handlers.put(DiscoveryMessageHandler.getMessageClass(), new DiscoveryMessageHandler(context));
        handlers.put(FindValueRequestHandler.getMessageClass(), new FindValueRequestHandler(context));
        handlers.put(StoreRequestHandler.getMessageClass(), new StoreRequestHandler(context));
        return handlers;
    }

}
